package com.yfbao.poll.demo.service;

import com.yfbao.poll.demo.event.PollingEvent;
import com.yfbao.poll.demo.message.BasePollMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.event.EventListener;
import org.springframework.core.annotation.Order;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;

/**
 * 轮询消息处理类
 */
@Component
public class PollingEventService {

    private static final Log logger = LogFactory.getLog(PollingEventService.class);
    //用户消息缓存，按照Map<user,Map<type,List<message>> 的方式存储
    private static final Map<String, List<BasePollMessage>> userMessageMap = PollingCache.getUserMessageCacheMap();

    //轮询业务逻辑处理类
    private static PollingService pollingService;

    public void setPollingService(PollingService pollingService) {
        this.pollingService = pollingService;
    }

    private static final Long messageTimeout = 3*60*1000L; // 消息再缓存中只保留3分钟

    static {
        //启动定时清理缓存的异步任务
        PollingExecutor.scheduleNormalTask(new DealyDelTimeoutMessageTask(),10,TimeUnit.SECONDS);
    }

    /**
     * 事件监听处理
     * @param event 消息
     */
    @Async("asyncTaskExecutor")
    @EventListener(condition = "")
    @Order(2)  //一个事件多个事监听，在同步的情况下，使用@order值越小，执行顺序优先
    public void pollingEventListen(PollingEvent event){
        logger.info(this.getClass().getSimpleName() + "监听到数据：" + event.getMsg());
        //1、先尝试直接推送，推送成功则不再保存至用户消息缓存中
        if(event ==null){
            logger.error("PollingEvent is null");
            return;
        }
        try {
            logger.info("revice PollingEvent:"+event.getMsg());
            BasePollMessage msg = event.getMsg();
            if(msg ==null){
                logger.error("PollingEvent msg is null");
                return;
            }
            boolean handlerRes = pollingService.messageHandlerTask(msg);
            if(handlerRes) {
                //启动一个清理历史消息的异步任务(标明处理了最新消息，将历史消息处理掉)
                PollingExecutor.executeNormalTask(new HistoryMessageHandlerTask(msg.getMsgId(),msg.getLoginName(),msg.getMsgType()));
                return;
            }

            //2、如果轮询队列中没有等待的轮询请求，则放入缓存中
            String loginName = msg.getLoginName();
            String msgType = msg.getMsgType();
            String cacheKey=loginName+"-"+msgType;
//            String cacheKey=loginName;
            List<BasePollMessage> userMessageList = userMessageMap.get(cacheKey);
            if(userMessageList == null){
                userMessageList =new CopyOnWriteArrayList<BasePollMessage>();
                List<BasePollMessage> basePollMessages = userMessageMap.putIfAbsent(cacheKey, userMessageList);
                if(basePollMessages !=null){
                    userMessageList = basePollMessages;
                }
            }
            userMessageList.add(msg);
            userMessageMap.put(cacheKey, userMessageList);
        }catch (Throwable t){
            logger.error("handler PollingEvent error",t);
        }
    }


    /**
     * 历史任务处理类
     * 将历史已处理的过消息从缓存队列中移除
     */
    class HistoryMessageHandlerTask implements Runnable {
        @Override
        public void run() {
            try {
                String cacheKey=this.userName+"-"+this.msgType;
//                String cacheKey=this.userName;
                List<BasePollMessage> userMessages = userMessageMap.get(cacheKey);
                if(userMessages!=null && userMessages.size()>0){
                    Iterator<BasePollMessage> iterator = userMessages.iterator();
                    while (iterator.hasNext()){
                        BasePollMessage message = iterator.next();
                        if(this.msgId !=null && message.getMsgId()>this.msgId){
                            continue;
                        }
                        //删除历史消息（当前处理的消息id已超过缓存中的id,标明缓存中消息已被处理过或为无效的历史消息，从缓存中删除）
//                        iterator.remove();
                        userMessages.remove(message);
                        logger.info("delete history msg:"+message.getLoginName()+","+message.getMsgType()+","+message.getMsgInfo());
                    }
                }
            }catch (Throwable t){
                logger.error("delete history error",t);
            }
        }

        private Long msgId;
        private String userName;
        private String msgType;
        HistoryMessageHandlerTask(Long msgId,String msgType,String userName){
            this.msgId=msgId;
            this.userName=userName;
            this.msgType=msgType;
        }
    }

    /**
     * 定时删除超期消息
     *  已定时任务的方式，定时删除超过设定超期时间的历史任务
     */
    static class DealyDelTimeoutMessageTask implements Runnable {

        @Override
        public void run() {
            try {
                Set<String> keySet = userMessageMap.keySet();
                if(keySet==null || keySet.isEmpty()){
                    return;
                }
                logger.info("DealyDelTimeoutMessageTask, keySet size:"+keySet.size());
                for (String key : keySet) {
                    List<BasePollMessage> userMessages = userMessageMap.get(key);
                    if(userMessages!=null && userMessages.size()>0){
                        logger.info("DealyDelTimeoutMessageTask,"+key+","+userMessages.size());
                        Iterator<BasePollMessage> iterator = userMessages.iterator();
                        while (iterator.hasNext()){
                            BasePollMessage message = iterator.next();
                            long currentTime = System.currentTimeMillis();
                            if((currentTime-message.getTimestamp()) < messageTimeout){
                                continue;
                            }
                            //删除历史消息（当前处理的消息id已超过缓存中的id,标明缓存中消息已被处理过或为无效的历史消息，从缓存中删除）
                            userMessages.remove(message);
                            logger.info("delete timeout pollMsg:"+message.getLoginName()+","+message.getMsgType()+","+message.getMsgInfo());
                        }
                    }
                }
            }catch (Throwable t){
                logger.error("delete timeout pollMsg error",t);
            }
        }
    }


}
